package que.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import que.app.AppV2;
import que.bean.TrafficPageViewBean;
import que.function.DimAsyncFunction;
import que.util.AtguiguUtil;
import que.util.FlinkSinkUtil;

import java.time.Duration;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;

import static que.common.Word.*;

/**
 * @author Naruto
 * @description:
 * @Class_Name que.app.dwd.log.Dws_VideoWatchWIndow
 * @Date 2022/09/01 16:22
 */
public class Dws_VideoWatchWIndow extends AppV2 {

    public static void main(String[] args) {
        new Dws_VideoWatchWIndow().init(
                9003,
                2,
                "Dws_VideoWatchWIndow",
                TOPIC_DWD_TRAFFIC_VIDEO, TOPIC_DWD_TRAFFIC_VIDEO_PLAY_INFO, TOPIC_DWD_TRAFFIC_UqViewer
        );
    }


    @Override
    protected void handle(StreamExecutionEnvironment env, HashMap<String, DataStreamSource<String>> stream) {

        //todo 1. 将三种流转成同一种类型, 然后union成一个流
        DataStream<TrafficPageViewBean> unionStream = parseAndUnionOne(stream);
//        unionStream.print();
        //补充维度
        SingleOutputStreamOperator<TrafficPageViewBean> dimStream = addDim(unionStream);
//        dimStream.print();

        //开窗聚合  ：因为聚合的key在补充的维度表中，所以先获取维度，再聚合
        SingleOutputStreamOperator<TrafficPageViewBean> result = windowandAgg(dimStream);

        //写出到doris中
        result.print();
//        writeToDoris(result);
//        result.print("main");
//        result.getSideOutput(new OutputTag<TrafficPageViewBean>("late"){}).print("late");

    }

    private void writeToDoris(SingleOutputStreamOperator<TrafficPageViewBean> result) {
        result
                .map(bean ->{
                    SerializeConfig conf = new SerializeConfig();
                    // 转成json的时候, 属性名使用下划线
                    conf.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;

                    String str = JSON.toJSONString(bean, conf);

                    return str;
                })
                .addSink(FlinkSinkUtil.getDorisSink("edu.dws_traffic_chaptor_video_watch_window"));
    }

    private SingleOutputStreamOperator<TrafficPageViewBean> addDim(DataStream<TrafficPageViewBean> unionStream) {

        // 1.补充章节维度信息
        SingleOutputStreamOperator<TrafficPageViewBean> dimStream = AsyncDataStream.unorderedWait(
                unionStream,
                new DimAsyncFunction<TrafficPageViewBean>() {
                    @Override
                    public String getTable() {
                        return "dim_chapter_info";
                    }

                    @Override
                    public String getId(TrafficPageViewBean input) {
                        return input.getVideoId();
                    }

                    @Override
                    public String getColunmName() {
                        return "video_id";
                    }

                    @Override
                    public void addDim(JSONObject dim, TrafficPageViewBean input) {
                        input.setChapterName(dim.getString("CHAPTER_NAME"));
                        input.setChapterId(dim.getString("ID"));
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        return dimStream;
    }

    private SingleOutputStreamOperator<TrafficPageViewBean> windowandAgg(DataStream<TrafficPageViewBean> unionStream) {
        SingleOutputStreamOperator<TrafficPageViewBean> aggStream = unionStream
                .filter(bean -> {
                    return bean.getChapterId() != null && bean.getChapterId().length() > 0;
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(60))
                                .withTimestampAssigner((bean, ts) -> bean.getTs())
                )
                .keyBy(bean -> bean.getChapterId())//根据章节分类
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sideOutputLateData(new OutputTag<TrafficPageViewBean>("late"){})
                .reduce(
                        new ReduceFunction<TrafficPageViewBean>() {
                            @Override
                            public TrafficPageViewBean reduce(TrafficPageViewBean bean1, TrafficPageViewBean bean2) throws Exception {
                                bean1.setSvCt(bean1.getSvCt() + bean2.getSvCt());//视频播放次数
                                bean1.setDurSum(bean1.getDurSum() + bean2.getDurSum());//累计播放时长
                                bean1.setUvCt(bean1.getUvCt() + bean2.getUvCt());//观看人数
                                return bean1;
                            }
                        },
                        new ProcessWindowFunction<TrafficPageViewBean, TrafficPageViewBean, String, TimeWindow>() {
                            @Override
                            public void process(String key,
                                                Context ctx,
                                                Iterable<TrafficPageViewBean> it,
                                                Collector<TrafficPageViewBean> out) throws Exception {
                                TrafficPageViewBean bean = it.iterator().next();

                                bean.setStt(AtguiguUtil.toDateTime(ctx.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(ctx.window().getEnd()));
                                bean.setCurDate(AtguiguUtil.toDateTime(System.currentTimeMillis()));//统计日期
                                out.collect(bean);
                            }
                        }
                );

        return aggStream;


    }

    private DataStream<TrafficPageViewBean> parseAndUnionOne(HashMap<String, DataStreamSource<String>> mapStreams) {

        //记录视频播放次数
        SingleOutputStreamOperator<TrafficPageViewBean> stream1 = mapStreams
                .get(TOPIC_DWD_TRAFFIC_VIDEO_PLAY_INFO)
                .map(str -> {
                    JSONObject json = JSONObject.parseObject(str);

                    Long svCt = 1l; //一条算一次播放 不算一个章节
                    Long uvCt = 0l;
                    Long durSum = 0l;

                    Long ts = json.getLong("ts");

                    return TrafficPageViewBean.builder()
                            .videoId(json.getJSONObject("appVideo").getString("video_id"))
                            .uvCt(uvCt)
                            .svCt(svCt)
                            .durSum(durSum)
                            .ts(ts)
                            .build();

                });

//        stream1.print("stream1");
        //播放用户数
        SingleOutputStreamOperator<TrafficPageViewBean> stream2 = mapStreams
                .get(TOPIC_DWD_TRAFFIC_UqViewer)
                .map(str -> {
                    JSONObject json = JSONObject.parseObject(str);

                    Long svCt = 0l;
                    Long uvCt = 1l;//一条算一一个播放用户
                    Long durSum = 0l;

                    Long ts = json.getLong("ts");

                    return TrafficPageViewBean.builder()
                            .videoId(json.getJSONObject("appVideo").getString("video_id"))
                            .uvCt(uvCt)
                            .svCt(svCt)
                            .durSum(durSum)
                            .ts(ts)
                            .build();
                });
//        stream2.print("stream2");
        //播放总时长
        SingleOutputStreamOperator<TrafficPageViewBean> stream3 = mapStreams
                .get(TOPIC_DWD_TRAFFIC_VIDEO)
                .map(str -> {
                    JSONObject json = JSONObject.parseObject(str);
                    JSONObject video = json.getJSONObject("appVideo");


                    Long svCt = 0l;
                    Long uvCt = 0l;
                    Long durSum = video.getLong("play_sec");//一条数据的播放时长

                    Long ts = json.getLong("ts");
//                    System.out.println("33333333 :"+ durSum);
                    return TrafficPageViewBean.builder()
                            .videoId(json.getJSONObject("appVideo").getString("video_id"))
                            .uvCt(uvCt)
                            .svCt(svCt)
                            .durSum(durSum)
                            .ts(ts)
                            .build();
                });
//        stream3.print("stream3");

         return stream1.union(stream2,stream3);
    }
}
